package jms;

import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;

import tapp.ActiveMQSessionTest;


public class MessageSender {
	final BlockingQueue<MessageSpec> messageQueue = new LinkedBlockingQueue<MessageSpec>(100);
	ExecutorService bossPool;
	ExecutorService workerPool;
	//final Session session;
	//final MessageProducer producer;
	//final SessionProducerPair spPair;
	//final MessageConsumer consumer;
	final Destination replyDest;
	final MessageDispatcherManager dispatcherManager;
	
	final BlockingQueue<SessionProducerPair> spPairQueue = 
			new LinkedBlockingQueue<SessionProducerPair>(100);
	
	static final int MAX_SENDER = 10;
	private final AtomicInteger workerCounter = new AtomicInteger();
	
	
	public MessageSender(Destination replyTo, MessageDispatcherManager dispatcherManager) {
		this.replyDest = replyTo;
		this.dispatcherManager = dispatcherManager;
	}
	
	public void addSessionProducerPair(Session s, MessageProducer p) {
		try {
			spPairQueue.put(new SessionProducerPair(s, p));
		} catch (InterruptedException ex) {
			return;
		}
	}
	
	public void addSessionProducerPair(SessionProducerPair pair) {
		try {
			spPairQueue.put(pair);
		} catch (InterruptedException ex) {
			return;
		}
	}
	
	public void start() {
		bossPool = Executors.newSingleThreadExecutor(new NamedThreadFactory("Boss"));
		workerPool = Executors.newCachedThreadPool(new NamedThreadFactory("Sender-Workder"));
		bossPool.execute(new Runnable() {
			public void run() {
				while(true) {
					final MessageSpec message;
					try {
						message = messageQueue.take();
					}catch(InterruptedException ex){
						break;
					}

					while(workerCounter.incrementAndGet() > MAX_SENDER) {
						workerCounter.decrementAndGet();
						try {
							Thread.sleep(100);
						} catch(InterruptedException ex) {
							return;	//stop bossPool
						}
					}
					
					workerPool.execute(new Runnable() {
						public void run() {
							try {
								sendMessage(message);
							} catch(Exception ex) {
								Logger.getAnonymousLogger().log(Level.SEVERE, "", ex);
							} finally {
								workerCounter.decrementAndGet();
							}
						}
					});
				}
			}
			
		});

	}

	public void sendMessage(MessageSpec spec) throws Exception {
		final SessionProducerPair spPair;
		try {
			spPair = spPairQueue.take();
		}catch(InterruptedException ex) {
			return;
		}
		
		try {
			Message message = spec.buildMessage(spPair.getSession());
			message.setJMSReplyTo(replyDest);
			String cId = UUID.randomUUID().toString();
			if(cId == null)
				throw new RuntimeException("cid");
			message.setJMSCorrelationID(cId);
			dispatcherManager.register(cId, spec.getDispatcher());
			
			spPair.getProducer().send(message);
			
		} finally {
			try {
				spPairQueue.put(spPair);
			} catch (InterruptedException ex) {
				return;
			}
		}
		//System.out.println("Message send => " + cId);
	}
	
	public void addMessage(MessageSpec message) {
		try {
			messageQueue.put(message);
			//System.out.println("Add message " + messageQueue.size());
		}catch(InterruptedException ex) {
			Thread.currentThread().interrupt();
		}
	}
	
	public void stop() {
		bossPool.shutdown();
		workerPool.shutdown();
		
		try {
			bossPool.awaitTermination(1, TimeUnit.SECONDS);
		}catch(InterruptedException ex) {
			Logger.getAnonymousLogger().warning("Interrupted");
			Thread.currentThread().interrupt();
		}
		
		try {
			workerPool.awaitTermination(10, TimeUnit.SECONDS);
		}catch(InterruptedException ex){
			Logger.getAnonymousLogger().warning("Interrupt");
		}
		
		bossPool.shutdownNow();
		workerPool.shutdownNow();
	}
}